package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class 各品牌支付额TOP10个商品 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt/");
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 订单主表
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        "      id BIGINT," +
                        "      status INT,                        " +
                        "      create_time timestamp(3),          " +
                        "      modify_time timestamp(3),          " +
                        "      payment_time timestamp(3),         " +
                        "      rt as create_time        ,         " +
                        "      watermark for rt as rt - interval '0' second ,  " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order'          " +
                        ")"
        );

        // 订单详情表
        tenv.executeSql(
                "CREATE TABLE item_mysql (    " +
                        "      id       BIGINT,  " +
                        "      order_id BIGINT,  " +
                        "      product_id BIGINT,                 " +
                        "      product_brand STRING,              " +
                        "      real_amount decimal(10,2),       " +
                        "     PRIMARY KEY (id) NOT ENFORCED       " +
                        "     ) WITH (                            " +
                        "     'connector' = 'mysql-cdc',          " +
                        "     'hostname' = 'doitedu'   ,          " +
                        "     'port' = '3306'          ,          " +
                        "     'username' = 'root'      ,          " +
                        "     'password' = 'root'      ,          " +
                        "     'database-name' = 'realtimedw',     " +
                        "     'table-name' = 'oms_order_item'     " +
                        ")"
        );


        // 关联
        tenv.executeSql(
                " CREATE TEMPORARY VIEW joined AS  SELECT  " +
                        "    o.id as order_id,                " +
                        "    o.create_time ,                  " +
                        "    o.payment_time,                  " +
                        "    o.status,                        " +
                        /*"    o.rt,                            " +*/
                        "    i.product_id,                    " +
                        "    i.product_brand,                 " +
                        "    i.real_amount                    " +
                        " from order_mysql o                  " +
                        " join item_mysql i                   " +
                        " on o.id = i.order_id                 "
        );

        // 关联后， 源表中声明的 事件时间语义字段 rt ，已经失去了时间语义属性，只是一个普通的timestamp而已了
        // 而后续的时间窗口聚合，必须要一个时间语义属性的字段
        // 我们可以在这里想办法重新定义一个 rt
        Table table1 = tenv.from("joined");
        DataStream<Row> changelogStream = tenv.toChangelogStream(table1);

        // 再将流转成表，在此过程中重新定义  rowtime 属性
        Table table2 = tenv.fromChangelogStream(changelogStream,
                Schema.newBuilder()
                        .column("order_id", DataTypes.BIGINT())
                        .column("create_time", DataTypes.TIMESTAMP(3))
                        .column("payment_time", DataTypes.TIMESTAMP(3))
                        .column("status", DataTypes.INT())
                        .column("product_id", DataTypes.BIGINT())
                        .column("product_brand", DataTypes.STRING())
                        .column("real_amount", DataTypes.DECIMAL(10, 2))
                        .columnByExpression("rt", "create_time")
                        // 重新定义rowtime字段
                        .watermark("rt", "rt - interval '0' second")
                        .build());
        tenv.createTemporaryView("joined2",table2);


        // 窗口聚合出 每个品牌金额最大的前10个商品
        tenv.executeSql(
                "SELECT\n" +
                        "    window_start,\n" +
                        "    window_end,\n" +
                        "    product_brand,\n" +
                        "    product_id,\n" +
                        "    real_amount,\n" +
                        "    row_number() over(PARTITION BY window_start,window_end,product_brand ORDER BY real_amount asc ) as rn \n" +
                        "FROM (\n" +
                        "    SELECT\n" +
                        "        TUMBLE_START(rt,INTERVAL '24' HOUR) as window_start,\n" +
                        "        TUMBLE_END(rt,INTERVAL '24' HOUR) as window_end,\n" +
                        "        TUMBLE_ROWTIME(rt,INTERVAL '24' HOUR) as rt,\n" +
                        "        product_brand,\n" +
                        "        product_id,\n" +
                        "        sum(real_amount) as real_amount \n" +
                        "    from joined2\n" +
                        "    GROUP BY \n" +
                        "        TUMBLE(rt,INTERVAL '24' HOUR),\n" +
                        "        product_brand,\n" +
                        "        product_id\n" +
                        ") o "

        ).print();


    }

}
